package com.grpc;

import hello.HelloRequest;
import hello.HelloResponse;
import hello.HelloResponseList;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;

public class HelloClient {

    private final ManagedChannel channel;
    private final HelloGrpc.HelloBlockingStub blockingStub;
    private final HelloGrpc.HelloStub stub;


    public HelloClient(String host, int port) {
        channel = ManagedChannelBuilder.forAddress(host, port)
                .usePlaintext(true)
                .build();

        blockingStub = HelloGrpc.newBlockingStub(channel);
        stub = HelloGrpc.newStub(channel);
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    //一元流
    public void greet(String name) {
        HelloRequest request = HelloRequest.newBuilder().setName(name).build();
        HelloResponse response = blockingStub.sayHello(request);
        System.out.println(request.getName());
        System.out.println(response.getMessage());
    }

    //服务流 ok
    public void serviceStream(String name) {
        HelloRequest request = HelloRequest.newBuilder().setName(name).build();

        Iterator<HelloResponse> iterator = blockingStub.serviceStreamHello(request);
        System.out.println(request.getName());
        while (iterator.hasNext()) {//服务端发送了多个流，这里循环接收
            HelloResponse response = iterator.next();
            System.out.println(response.getMessage());
        }
    }

    //客户流
    public void clientStream(String name) {
        StreamObserver<HelloResponseList> helloResponseList = new StreamObserver<HelloResponseList>() {

            @Override
            public void onNext(HelloResponseList helloResponseList) {
                helloResponseList.getHelloResponseList().forEach(e -> {
                    System.out.println(e.getMessage());
                });
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("completed!");
            }
        };

        StreamObserver<HelloRequest> helloRequestStreamObserver = stub.clientStreamHello(helloResponseList);
        helloRequestStreamObserver.onNext(HelloRequest.newBuilder().setName("客户端:小红好").build());
        helloRequestStreamObserver.onNext(HelloRequest.newBuilder().setName("客户端:吃饭了吗").build());
        helloRequestStreamObserver.onNext(HelloRequest.newBuilder().setName("客户端:今天出去玩吗").build());
        helloRequestStreamObserver.onCompleted();

        try {
            //因为grpc调用参数为stream的时候，都是采取异步处理，客户端程序不会等待结果响应
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //客户服务流
    public void clientServerStream(String name) {

        StreamObserver<HelloResponse> helloResponse = new StreamObserver<HelloResponse>() {
            @Override
            public void onNext(HelloResponse helloResponse) {
                System.out.println(helloResponse.getMessage());
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("completed!");
            }
        };

        StreamObserver<HelloRequest> helloRequestStreamObserver = stub.clientServiceStreamHello(helloResponse);

        for (int i = 0; i < 10; i++) {
            helloRequestStreamObserver.onNext(HelloRequest.newBuilder().setName(name + i).build());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        try {
            //因为grpc调用参数为stream的时候，都是采取异步处理，客户端程序不会等待结果响应
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) throws InterruptedException {
        HelloClient client = new HelloClient("127.0.0.1", 50051);

//        client.clientStream("客户端流参数");
      client.clientServerStream("客户服务流参数");


        //      client.greet("客户端:小红好");
        //client.serviceStream("客户端:小红好");
    }
}